#include <cassert>
#include <cstring>
#include <fstream>
#include <netdb.h>
#include <set>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/types.h>

#include "../../Error.h"
#include "../../tools/Log.h"
#include "../../tools/Time.h"
#include "../../tools/network/SelectLoop.h"

#include "../commands/GameCommand.h"
#include "../GameControl.h"

#include "GameNetwork.h"
#include "GamePacket.h"

using namespace std;

GameNetwork::GameNetwork(unsigned long initial_ok_timestamp) :
	run(true), network_frame_num(0), last_command_flush(0), current_update(), pending_acks(), network_thread(),
	sock(-1), neighbors(), ok_timestamp(initial_ok_timestamp)
{ }

GameNetwork::~GameNetwork()
{
	if(network_thread)
	{
		void* tmp;
		run = false;
		CHECK(pthread_join(network_thread, &tmp));
		network_thread = NULL;
	}
	
	close(sock);
}

/*	Must be called AFTER GameControl is instantiated because the network thread immediately looks to access the
	GameControl instance
*/
void GameNetwork::init(const NetworkConf& conf)
{
	unsigned initial_ok_time = GameControl::Instance()->get_command_horizon();
	
	for (unsigned i = 0; i < conf.players.size(); i ++)
	{
		if (conf.players[i] != conf.me)
			neighbors.push_back(new GameNetwork::Node(conf.players[i]->addr, initial_ok_time));
		else
			sock = init_socket(conf.players[i]->addr.sin_port);		// Found ourself => initialize our port
	}
	
	hosting = conf.i_am_host;	// Whether or not we're hosting the game
	
	CHECK(pthread_create(&network_thread, NULL, network_thread_main, this));
}

bool GameNetwork::i_am_host()
{
	return hosting;
}

GameNetwork::Node *GameNetwork::find_peer_by_addr(struct sockaddr *from)
{
	for (unsigned i = 0; i < neighbors.size(); i ++)
		if((neighbors[i]->addr.sin_addr.s_addr == ((struct sockaddr_in *) from)->sin_addr.s_addr) &&
		   (neighbors[i]->addr.sin_port == ((struct sockaddr_in *) from)->sin_port))
			return neighbors[i];
	
	return NULL;
}

// Creates and binds a UDP socket on port (port is expected in network byte order). Returns 0 if binding fails.
int GameNetwork::init_socket(int port)
{
	struct sockaddr_in myaddr;
	int sock;
	
	if ((sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_IP)) == -1) {
		lout << "WARNING: NetworkSystem - Unable to create socket." << endl;
		return 0;
	}
	
	bzero(&myaddr, sizeof(myaddr));
	myaddr.sin_family = AF_INET;
	myaddr.sin_addr.s_addr = htonl(INADDR_ANY);
	myaddr.sin_port = port;
	
	if (bind(sock, (struct sockaddr *) &myaddr, sizeof(myaddr)) == -1) {
		lout << "WARNING: NetworkSystem - Unable to bind socket." << endl;
		return 0;
	}
	
	return sock;
}


void* GameNetwork::network_thread_main(void* data)
{
	GameNetwork* network = (GameNetwork*) data;
	network->main_loop();
	
	return NULL;
}

using namespace GamePacket;

void GameNetwork::main_loop()
{
	class UpdateHandler : public SocketHandler
	{
		GameNetwork *network;
		
	public:
		UpdateHandler(GameNetwork *n) : network(n) { }
		UpdateHandler(const UpdateHandler&):SocketHandler(), network(NULL) {}
		UpdateHandler& operator=(const UpdateHandler&) { return *this; }
		
		virtual void operator()(void* buf, struct sockaddr* addr)
		{
			Header header = Header(buf);
			GameNetwork::Node *sender = network->find_peer_by_addr(addr);
			
			if (!sender) {
				lout << "WARNING: Network thread received packet from bogus peer." << endl;
				return;
			}
			
			if (header.type == ACK && header.seq_num == network->network_frame_num)		// Process ACK for last batch of commands
			{
				network->pending_acks.erase(sender);
				
				if (network->pending_acks.size() == 0)
					network->begin_next_frame();
			}
			
			else if (header.type == UPDATE)		// Process UPDATE
			{
				Ack ack = Ack(header.seq_num);
				ack.convert_and_send(network->sock, &sender->addr);		// ACK this UPDATE
				
				Update update = Update(buf);
				
				if (header.seq_num > sender->last_seq_num)		// If this batch of commands is new, queue them up
				{
					for (unsigned i = 0; i < update.data.num_commands; i ++)
					{
						GameCommand *cmd = GameCommand::interpret(update.data.commands[i]);
						assert(cmd != NULL);
						GameControl::Instance()->command_queue.push(cmd);
					}
					
					sender->last_seq_num = header.seq_num;
					sender->last_ok_time = update.data.ok_time;
					
					network->update_ok_timestamp();		// Increase Network's ok_timestamp if this was the slowest peer
				}
			}
		}
	};
	
	class UpdateTimeoutHandler : public TimeoutHandler
	{
		GameNetwork *network;
	
	public:
		UpdateTimeoutHandler(GameNetwork *n) : network(n) { }
		UpdateTimeoutHandler(const UpdateTimeoutHandler&): TimeoutHandler(), network(NULL) {}
		UpdateTimeoutHandler& operator=(const UpdateTimeoutHandler&) { return *this; }
		virtual ~UpdateTimeoutHandler()	{}
		
		virtual void operator()()
		{
			for (set<GameNetwork::Node*>::iterator itr = network->pending_acks.begin(); itr != network->pending_acks.end(); itr ++)
				network->current_update.convert_and_send(network->sock, &(*itr)->addr);
		}
	};
	
	socket_handler = loop.add_socket(sock, GamePacket::MAX_PACKET_LEN, new UpdateHandler(this));
	wakeup_handler = loop.add_wakeup_interval(GamePacket::PACKET_TIMEOUT, new UpdateTimeoutHandler(this));
	begin_next_frame();
	
	while (run)
		loop.run();
}

// Blocks until a batch of commands is available from Game
void GameNetwork::begin_next_frame()
{
	GameControl *g = GameControl::Instance();
	unsigned long our_ok_time;
	
	/*	We were about to send out an update with a dt of 0, which means we're waiting on Game to advance to its next frame.
		It's possible the reason Game can't advance to the next frame is that it was waiting on Network (us) to process more
		updates from our peers. Thus we need to wait on Game, but we can't simply block because it's our thread that Game
		could be waiting for (deadlock). We have to continue processing updates for game until it's ready to let us go again.
	*/
	while (run && (our_ok_time = g->get_command_horizon()) == last_command_flush)
		loop.run();
	
	for (unsigned i = 0; i < neighbors.size(); i ++)	// We need to hear ACKs from each neighbor
		pending_acks.insert(neighbors[i]);
	
	loop.reset_wakeup_interval(wakeup_handler);			// To avoid immediate double send
	
	network_frame_num ++;
	current_update = GamePacket::Update(our_ok_time, network_frame_num);	// Prepare our next update
	
	for (unsigned i = 0; i < g->command_buffer.size(); i ++)
	{
		if (g->command_buffer[i]->timestamp < g->get_command_horizon())
			current_update.add_command(*(g->command_buffer[i]));
		else
			break;
	}
	
	g->command_buffer.erase(g->command_buffer.begin(), g->command_buffer.begin() + current_update.data.num_commands);
	
	last_command_flush = our_ok_time;
	
	// Send new update to neighbors
	for (set<GameNetwork::Node*>::iterator itr = pending_acks.begin(); itr != pending_acks.end(); itr ++)
		current_update.convert_and_send(sock, &(*itr)->addr);
}

void GameNetwork::update_ok_timestamp()
{
	unsigned long new_network_ok_time = -1;		// Infinity
	
	for (unsigned i = 0; i < neighbors.size(); i ++)
		if (neighbors[i]->last_ok_time < new_network_ok_time)		// Find minimum ok_time among our peers
			new_network_ok_time = neighbors[i]->last_ok_time;
	
	// Even though ok_timestamp is shared between Game and Network threads, this doesn't need a mutex (atomic operation)
	ok_timestamp = new_network_ok_time;
}
